【Claude Code源码剖析】番外-两个月冲刺手册 — DeepSeek Agent Infra 面试无懈可击

目标：两个月后，面对任何 Agent Infra / LLM Application 面试题，能给出有深度、有源码支撑、有实战经历的完整回答。

基准：你的对手是有 2 年 Agent 工程经验的候选人。你要在两个月内追平他们，并在系统理解深度上超过他们。

---

战略原则（先读这里）

原则 1：理解 > 记忆
每个系统手写一遍，带着问题写，不是抄代码。写完问自己：”如果这里不这样设计，会出什么问题？”

原则 2：有作品 > 有文档
Week 7 的 Capstone 项目是最重要的。面试时”我做过 X”比”我知道 X”价值高 5 倍。

原则 3：Python 第一
DeepSeek 岗位考 Python。TypeScript 文档用来理解 CC 的设计思路，不需要手写 TS 代码。

原则 4：每天结束写三行
今天真正理解了什么（不是”读了什么”）。不能写出来，就是没懂。

---

每日执行模板

08:00–10:00   精读（文档 + 源码，做笔记）
10:00–12:00   手写代码（不看文档，凭理解）
13:00–15:00   调试 + 迭代（让代码真正跑起来）
15:00–16:30   扩展（改造、加功能、压测）
16:30–17:00   总结三行日志 + 更新进度表

每周日晚：做本周验收测试（见各周末”验收标准”）

---

Week 0（可选）：语法基础自检

是否需要 Week 0？ 对照下表自测，全部能答出则直接进 Week 1。

Python 自检（5 分钟，不看文档）

# 以下代码能流畅写出吗？（不是「大概会」，而是「闭眼能写」）

# 1. 用列表推导式 + 字典推导式各写一行
# 2. 用 @dataclass 定义一个带默认值字段的类
# 3. 写一个 context manager（__enter__ / __exit__ 或 @contextmanager）
# 4. 用 *args / **kwargs 写一个装饰器
# 5. typing 模块：Optional / Union / TypeVar / Generic 各怎么用？

答不出其中 2 题以上 → 先花 2 天学语法，再进 Week 1：

• Day 1：docs/tech-stack/python/syntax/01-基础入门.md
• Day 2：docs/tech-stack/python/syntax/02-进阶特性.md
• Day 3（选读）：docs/tech-stack/python/syntax/03-高阶精通.md

LLM 理论自检（快速判断）

能否用一句话解释：Transformer 的 Attention 计算、RLHF 的三步骤？

不能 → 花半天读：

• docs/LLM技术深度系列/01-Transformer架构完全深度解析.md（重点看 Self-Attention 章节）
• docs/LLM技术深度系列/03-RLHF完整技术解析.md（重点看三阶段流程）

TypeScript 自检（用于读 CC 源码）

不需要会写 TS，但要能读懂以下类型签名：

type Extract<T, U> = T extends U ? T : never
type AsyncGenerator<T, TReturn, TNext>
function buildTool<T extends ZodSchema>(config: ToolConfig<T>): Tool<z.infer<T>>

看不懂 → 花 1 天过：

• docs/tech-stack/typescript/syntax/01-基础入门.md（重点看泛型部分）
• docs/tech-stack/typescript/syntax/02-进阶特性.md（重点看条件类型）

---

Week 1：Python 异步 + 最小 Agentic Loop

学习材料

• docs/tech-stack/python/01-异步编程asyncio.md ← Day 1 精读
• docs/tech-stack/llm/01-AnthropicSDK与LLM接入.md ← Day 2 精读
• docs/02-Agentic查询循环与QueryEngine.md ← Day 3 理解架构

每天任务

Day 1：asyncio 基础

• 精读 python/01-异步编程asyncio.md
• 手写（不看文档）：

  # 独立完成以下 4 个实现
  # 1. asyncio.gather 并发执行 3 个 API 调用
  # 2. AsyncGenerator 流式输出（yield 每个 token）
  # 3. asyncio.timeout 给 API 调用加超时
  # 4. asyncio.Semaphore 限制并发数为 3

• 验证：4 个实现都能独立运行，不报错

Day 2：Anthropic SDK 接入

• 精读 llm/01-AnthropicSDK与LLM接入.md
• 手写（不看文档）：

  # 独立完成以下实现
  # 1. 非流式单轮对话
  # 2. 流式对话（逐 token 打印）
  # 3. 多轮对话（messages 列表追加）
  # 4. 带 Tool 定义的对话（先不处理工具调用，只看响应格式）

Day 3：理解 Agentic Loop 架构

• 精读 docs/02-Agentic查询循环与QueryEngine.md
• 在纸上画出完整的 Loop 流程图（从用户输入到最终响应）
• 关键问题：Loop 什么时候终止？怎么处理多个并发工具调用？

Day 4：手写 Agentic Loop v1

# 目标：不看任何参考，独立实现以下功能
# - 支持 Tool Use（bash、file_read 两个工具）
# - 流式输出文本
# - 工具调用后继续 Loop
# - Ctrl+C 正常退出（不崩溃）

async def agentic_loop(prompt: str) -> str:
    # 你来实现
    pass

不要追求完美，先让它跑起来。

Day 5：加入 withRetry

• 精读 docs/tech-stack/typescript/09-错误处理与重试策略.md
• 手写 Python 版 with_retry：

  # 要求：
  # - 指数退避：500ms × 2^(attempt-1)，上限 32s
  # - 25% 随机抖动
  # - 优先使用 Retry-After 响应头
  # - 429 和 529 都重试，其他 4xx 不重试
  # - 最多重试 10 次

Day 6：AbortController / 取消机制

• 精读 docs/tech-stack/typescript/02-异步编程与流式处理.md 中 AbortController 部分
• 手写 Python 版取消机制：

  # 要求：
  # - asyncio.Event 或 asyncio.CancelledError 实现取消
  # - 取消信号能传播到子 Agent 和工具调用
  # - Ctrl+C 发出取消信号，等待当前工具完成后退出（不强杀）

Day 7：验收测试

下面的代码，闭眼写出来（不看文档），5 分钟内写完：

# 验收：实现一个支持流式输出 + 工具调用 + 取消的最小 Agent
# 如果超过 30 分钟还没写出来 → 重新理解，明天再试

口述以下问题（对着镜子说）：

• Agentic Loop 为什么要用 while True？什么时候 break？
• 工具调用的 ID（tool_use_id）有什么作用？
• 429 和 529 的区别，为什么后台请求遇到 529 要直接丢弃？

---

Week 2：Tool System 深度 + 权限

学习材料

• docs/tech-stack/typescript/06-工具系统完整实现.md
• docs/tech-stack/typescript/04-安全与权限模式.md
• docs/03-工具系统Tool-System.md

每天任务

Day 1：Tool 接口设计

• 精读 typescript/06-工具系统完整实现.md
• 用 Python 实现 Tool 基类：

  from abc import ABC, abstractmethod
  from pydantic import BaseModel
  
  class Tool(ABC):
      name: str
      description: str
  
      @abstractmethod
      def input_schema(self) -> dict:  # JSON Schema
          pass
  
      @abstractmethod
      async def execute(self, input: dict) -> str:
          pass
  
      def validate_input(self, input: dict) -> bool:
          # 用 jsonschema 校验
          pass

Day 2：实现 BashTool

# 要求：
# - 真正执行 Shell 命令（subprocess）
# - 超时控制（默认 120s）
# - 危险命令拦截：rm -rf /、chmod 777、curl | bash 等
# - stdout + stderr 都要捕获
# - 非零退出码要报错（包含退出码和 stderr）
# - 工作目录可配置

Day 3：实现 FileReadTool + FileWriteTool

# FileReadTool：
# - 支持行号范围（startLine, endLine）
# - 大文件截断（> 10MB 拒绝）
# - 返回带行号的内容

# FileWriteTool：
# - 原子写（写临时文件，rename 到目标）
# - 目录不存在时自动创建
# - 记录写入的字节数

Day 4：三态权限系统

• 精读 typescript/04-安全与权限模式.md
• 手写 Python 版权限系统：

  from enum import Enum
  
  class PermissionResult(Enum):
      ALLOW = "allow"
      DENY = "deny"
      ASK = "ask"
  
  class PermissionChecker:
      def check(self, tool_name: str, input: dict) -> PermissionResult:
          # 1. 检查 allow_list（精确匹配 + 通配符）
          # 2. 检查 deny_list（优先级高于 allow）
          # 3. 检查危险模式（正则）
          # 4. 默认 ASK
          pass

Day 5：路径安全检查

# 要求实现以下安全检查
def is_path_safe(path: str, allowed_dirs: list[str]) -> bool:
    # 1. 解析绝对路径（防止 ../../../etc/passwd）
    # 2. 必须在 allowed_dirs 之一的子目录内
    # 3. 不允许访问 .git/config、~/.ssh/ 等敏感路径

# 手写测试：
assert is_path_safe("/tmp/work/file.txt", ["/tmp/work"]) == True
assert is_path_safe("/tmp/work/../../etc/passwd", ["/tmp/work"]) == False
assert is_path_safe("/home/user/.ssh/id_rsa", ["/tmp/work"]) == False

Day 6：工具注册表 + 动态加载

# 实现工具注册表，支持：
# - 按名称注册/查找工具
# - 从工具列表生成 Anthropic API 的 tools 参数格式
# - 懒加载（只在真正需要时初始化工具）

class ToolRegistry:
    def register(self, tool: Tool) -> None: ...
    def get(self, name: str) -> Tool | None: ...
    def to_api_format(self) -> list[dict]: ...  # Anthropic tools 参数格式

Day 7：验收测试

口述（不看笔记，2 分钟内说清楚）：

• CC 的 buildTool 工厂函数做了什么？为什么要有 TOOL_DEFAULTS？
• 危险命令检测用什么数据结构？时间复杂度是多少？
• 权限三态（allow/deny/ask）和 Linux 的权限系统有什么区别？

编码题（10 分钟内完成）：

# 实现一个函数，判断 bash 命令是否包含注入风险
# 危险模式：$() 命令替换、${} 变量展开、<() 进程替换、; && || 链式命令
def is_dangerous_command(cmd: str) -> bool:
    pass

# 测试用例
assert is_dangerous_command("ls -la") == False
assert is_dangerous_command("echo $(cat /etc/passwd)") == True
assert is_dangerous_command("rm -rf / --no-preserve-root") == True

---

Week 3：MCP 协议（DeepSeek 面试必考）

学习材料

• docs/tech-stack/llm/02-MCP协议实践.md
• docs/09-MCP协议集成.md

每天任务

Day 1：MCP 协议理解

• 精读 llm/02-MCP协议实践.md 全文
• 在纸上画：MCP Client ↔ JSON-RPC ↔ MCP Server 通信图
• 理解：Resources / Tools / Prompts 三类能力的区别
• 理解：stdio transport vs SSE transport 的适用场景

Day 2：手写最小 MCP Server（stdio）

# 要求：实现一个 stdio 传输的 MCP Server
# 支持以下方法：
# - initialize：返回 server info + capabilities
# - tools/list：列出可用工具
# - tools/call：执行工具调用
# 工具：file_read、bash_exec（简化版）

# 测试：用 echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{...}}' | python server.py
# 能返回正确的 JSON-RPC 响应

Day 3：手写 MCP Client

# 要求：实现 MCP Client，连接到你的 Server
# - 启动 Server 进程（subprocess）
# - 通过 stdin/stdout 发送 JSON-RPC 消息
# - 支持 initialize → tools/list → tools/call 完整流程
# - 处理异步响应（用 id 匹配请求和响应）

class MCPClient:
    async def connect(self, server_cmd: list[str]) -> None: ...
    async def list_tools(self) -> list[dict]: ...
    async def call_tool(self, name: str, arguments: dict) -> str: ...
    async def close(self) -> None: ...

Day 4：MCP 集成到 Agentic Loop

# 修改你的 Agentic Loop：
# - 启动时连接 MCP Server（可配置多个）
# - MCP Server 的工具与本地工具合并注册
# - 工具调用时区分本地工具和 MCP 工具
# - MCP Server 崩溃时优雅降级（该工具不可用，其他继续）

Day 5：SSE Transport 实现

# 实现 SSE 传输的 MCP Server（比 stdio 更适合远程部署）
# 要求：
# - /sse 端点：建立 SSE 连接，接收服务器推送
# - /messages 端点：POST JSON-RPC 请求
# - 用 FastAPI + anyio 实现

from fastapi import FastAPI
from sse_starlette import EventSourceResponse

app = FastAPI()

@app.get("/sse")
async def sse_endpoint(request: Request):
    # 实现 SSE 连接
    pass

@app.post("/messages")
async def handle_message(request: Request):
    # 处理 JSON-RPC 请求
    pass

Day 6：Resources 和 Prompts

# 扩展你的 MCP Server，支持：
# Resources：暴露文件系统资源（按 URI 读取）
#   resources/list → 列出可用资源
#   resources/read → 读取资源内容

# Prompts：预定义的 Prompt 模板
#   prompts/list → 列出可用模板
#   prompts/get  → 获取填充后的 Prompt

Day 7：验收测试

口述（2 分钟，不看笔记）：

• JSON-RPC 2.0 请求格式？必须有哪些字段？notification（通知）和普通请求的区别？
• MCP 的 initialize 握手为什么必须是第一个请求？
• stdio transport 和 SSE transport 各适合什么场景？为什么 CC 在本地用 stdio？

系统设计题（10 分钟口述）：

如果你要为一个 Agent 系统设计”工具市场”（类似 npm，用户可以发布/安装工具），基于 MCP 协议，你会怎么设计？

---

Week 4：Session 持久化 + Prompt 工程

学习材料

• docs/tech-stack/nodejs/03-Session持久化与JSONL.md
• docs/tech-stack/llm/03-Prompt工程与SystemPrompt.md
• docs/tech-stack/llm/04-上下文窗口管理与压缩.md
• docs/22-Session管理与状态恢复.md

每天任务

Day 1：JSONL 持久化

# 实现 Session 存储（读你的 JSONL 文档后手写）
# 数据结构：
# {"type": "user", "content": "...", "timestamp": "...", "id": "uuid"}
# {"type": "assistant", "content": [...], "id": "uuid", "usage": {...}}
# {"type": "tool_result", "tool_use_id": "...", "content": "..."}
# {"type": "compact_boundary", "summary": "...", "discarded_count": 10}

class SessionStore:
    async def append(self, message: dict) -> None:
        # 原子追加（先写临时文件，rename）
        pass

    async def load(self, session_id: str) -> list[dict]:
        # 从 compact_boundary 之后开始加载（丢弃已压缩的历史）
        pass

    async def list_sessions(self) -> list[dict]:
        # 列出所有 session（id + 首行 + 最后更新时间）
        pass

Day 2：Session 恢复

# 实现：从磁盘恢复上一次对话
# 要求：
# - 找到最新的 compact_boundary，从该点之后加载消息
# - 将已压缩的历史作为 System Prompt 的一部分注入
# - 恢复后的 Agent 能继续上下文提问

async def restore_session(session_id: str) -> tuple[str, list[dict]]:
    # 返回 (summary_for_system_prompt, messages_after_compact_boundary)
    pass

Day 3：五层 SystemPrompt 构建

• 精读 llm/03-Prompt工程与SystemPrompt.md
• 手写 Python 版 build_system_prompt：

  def build_system_prompt(
      base: str,           # 基础 System Prompt
      tools: list[Tool],   # 工具描述
      memory: str,         # CLAUDE.md 内容
      context: str,        # 当前工作目录等动态上下文
      override: str | None = None,  # 用户覆盖（优先级最高）
  ) -> str:
      # 5 个层级，按优先级合并
      # 注意 Prompt Cache 分段点的位置
      pass

Day 4：Prompt Cache 实战

# 目标：实测 Prompt Cache 节省效果
# 步骤：
# 1. 构建一个 5000 token 的固定 System Prompt
# 2. 第一次调用：记录 cache_creation_input_tokens
# 3. 第二次调用（相同 System Prompt）：记录 cache_read_input_tokens
# 4. 计算节省率：cache_read / (cache_read + input) ≈ 目标 90%+

# 手写工具：计算 Prompt Cache 节省金额
def calculate_cache_savings(usage: dict, model: str) -> float:
    # 基于 modelCost.ts 中的定价常量
    # cache_read 价格 = input_tokens 价格 × 0.1（省 90%）
    pass

Day 5：上下文压缩

• 精读 llm/04-上下文窗口管理与压缩.md
• 手写自动压缩逻辑：

  AUTOCOMPACT_THRESHOLD = 0.85  # 窗口使用 85% 触发压缩
  
  async def maybe_compact(
      messages: list[dict],
      current_tokens: int,
      max_tokens: int,
  ) -> list[dict]:
      if current_tokens / max_tokens < AUTOCOMPACT_THRESHOLD:
          return messages  # 不需要压缩
  
      # 调用 LLM 生成历史摘要
      summary = await generate_summary(messages)
  
      # 插入 compact_boundary 标记，丢弃旧消息
      return [
          {"type": "compact_boundary", "summary": summary},
          # 保留最近 N 条消息
          *messages[-20:],
      ]

Day 6：Token 计数实战

# 实现离线 token 估算（不调 API）
def estimate_tokens(text: str, file_extension: str = "") -> int:
    # JSON/JSONL：chars / 2
    # 其他文本：chars / 4
    # 中文：chars / 1.5
    pass

# 实现在线精确计数（调 Haiku，成本最低）
async def count_tokens_exact(messages: list, tools: list) -> int:
    # 使用 anthropic.beta.messages.count_tokens()
    pass

Day 7：验收测试

系统设计（15 分钟口述 + 画图）：

设计一个 Agent 的上下文管理系统，要求：

• 支持 200k token 的上下文窗口
• 超限时自动压缩，保证关键信息不丢失
• 重启后能完整恢复会话
• 多用户并发时数据不互相干扰

---

Week 5：多 Agent 协调

学习材料

• docs/tech-stack/typescript/07-多Agent协调系统.md
• docs/11-任务与多Agent系统.md
• docs/20-Swarm多Agent协调系统.md

每天任务

Day 1–2：Sub-Agent 生命周期

# 实现 SubAgent（有独立上下文 + 工具限制）
class SubAgent:
    def __init__(
        self,
        tools: list[str],         # 只允许使用的工具名称
        system_prompt: str,       # 独立于 parent 的 System Prompt
        max_turns: int = 20,      # 最大轮次限制
    ):
        self._messages: list[dict] = []   # 独立消息历史
        self._abort: asyncio.Event = asyncio.Event()

    async def run(self, task: str) -> str:
        # 完整的 Agentic Loop（隔离于 parent）
        pass

    def abort(self) -> None:
        self._abort.set()  # 级联取消

Day 3：Fork 模式 vs 普通子 Agent

# Fork：共享 parent 的消息历史（Prompt Cache 复用）
# 普通：完全独立上下文

class ForkedAgent(SubAgent):
    def __init__(self, parent_messages: list[dict], **kwargs):
        super().__init__(**kwargs)
        # Fork 时复制 parent 的历史作为只读上下文
        self._messages = parent_messages.copy() + []  # 浅拷贝，不修改 parent

# 关键：Fork 的 messages 前缀与 parent 相同
# → Prompt Cache 命中率接近 100%（省钱，省时间）

Day 4–5：Coordinator 模式（Swarm）

# Orchestrator 分配任务给多个 Worker Agent
# 要求：
# - Orchestrator 接收用户任务，分解为子任务
# - 将子任务分配给 Worker Agent（可并发）
# - Worker 完成后，结果汇总给 Orchestrator
# - Orchestrator 生成最终响应

async def coordinator(task: str, num_workers: int = 3) -> str:
    # 1. Orchestrator 分解任务
    subtasks = await decompose_task(task, num_workers)

    # 2. 并发运行 Worker Agent
    results = await asyncio.gather(*[
        SubAgent(tools=["bash", "file_read"]).run(subtask)
        for subtask in subtasks
    ])

    # 3. Orchestrator 汇总
    return await synthesize_results(task, results)

Day 6：错误隔离 + 级联取消

# 要求：
# - Worker Agent 崩溃不影响其他 Worker
# - 主 Agent 发出取消时，所有 Worker 在完成当前工具后停止
# - 超时的 Worker 不阻塞整体结果

async def run_with_isolation(agents: list[SubAgent]) -> list[str | Exception]:
    results = await asyncio.gather(
        *[agent.run(task) for agent in agents],
        return_exceptions=True,  # 单个失败不影响其他
    )
    return results

Day 7：验收测试

系统设计（20 分钟）：

设计一个”代码审查 Agent”：

• 接收 PR 的 diff
• 并行分配给 3 个子 Agent：安全审查、性能审查、代码风格审查
• 汇总结果，生成综合审查报告
  画出架构图，说明通信协议和错误处理

---

Week 6：Python 生态 + 算法 + RAG

学习材料

• docs/tech-stack/python/02-Pydantic数据校验.md
• docs/tech-stack/python/03-LLM生态对接.md
• docs/tech-stack/python/05-RAG检索增强生成.md
• docs/tech-stack/algorithms/02-LLM-Infra面试高频算法.md

每天任务

Day 1：Pydantic + Tool Schema

# 用 Pydantic 定义工具的输入 Schema
from pydantic import BaseModel, Field

class BashToolInput(BaseModel):
    command: str = Field(..., description="Shell command to execute")
    working_dir: str = Field("/tmp", description="Working directory")
    timeout: int = Field(120, ge=1, le=600, description="Timeout in seconds")

    model_config = {"extra": "forbid"}  # 拒绝未知字段

# 自动生成 JSON Schema（对接 Anthropic tools 格式）
schema = BashToolInput.model_json_schema()

Day 2：DeepSeek API 对接

# DeepSeek 兼容 OpenAI SDK，但有差异
from openai import OpenAI

client = OpenAI(
    api_key="your-deepseek-key",
    base_url="https://api.deepseek.com",
)

# 注意：DeepSeek 的 Reasoner 模型有 reasoning_content 字段
# 注意：Tool Use 格式与 OpenAI 完全相同
# 注意：流式响应处理方式与 OpenAI 相同

# 手写：DeepSeek + Tool Use 的完整 Agentic Loop

Day 3：RAG Pipeline 基础

# 手写完整 RAG Pipeline（不用 LangChain）
import chromadb
from sentence_transformers import SentenceTransformer

class RAGPipeline:
    def __init__(self):
        self.encoder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
        self.db = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.db.get_or_create_collection("docs")

    def index(self, docs: list[str], ids: list[str]) -> None:
        embeddings = self.encoder.encode(docs).tolist()
        self.collection.add(documents=docs, embeddings=embeddings, ids=ids)

    def search(self, query: str, top_k: int = 5) -> list[str]:
        q_emb = self.encoder.encode([query]).tolist()
        results = self.collection.query(query_embeddings=q_emb, n_results=top_k)
        return results["documents"][0]

    def ask(self, question: str) -> str:
        context = "\n".join(self.search(question))
        # 调 LLM 生成答案
        pass

Day 4：混合检索（Hybrid Search）

# 向量检索 + BM25 关键词检索，用 RRF 融合
from rank_bm25 import BM25Okapi

def rrf_fusion(
    vector_results: list[str],
    bm25_results: list[str],
    k: int = 60,
) -> list[str]:
    scores: dict[str, float] = {}
    for rank, doc in enumerate(vector_results):
        scores[doc] = scores.get(doc, 0) + 1 / (k + rank + 1)
    for rank, doc in enumerate(bm25_results):
        scores[doc] = scores.get(doc, 0) + 1 / (k + rank + 1)
    return sorted(scores, key=scores.get, reverse=True)

Day 5：算法题（每题独立实现）

题目	要求	时间限制
LRU Cache（LeetCode 146）	双向链表 + HashMap	15 分钟
令牌桶限流器	O(1) consume，支持突发	20 分钟
滑动窗口限流	O(N) 精确计量	15 分钟
Top-K（堆）	不全排序，O(N log K)	10 分钟
一致性哈希	150 个虚拟节点	20 分钟

Day 6：算法题复习

• 把 Day 5 每道题不看答案再写一遍
• 写完对比：有没有 bug？边界条件处理了吗？

Day 7：验收测试

算法题（面试节奏，20 分钟内完成）：

# 设计一个 API Gateway 的限流系统
# 要求：
# - 用户级别：每分钟 100k tokens（滑动窗口）
# - 全局级别：每秒 10000 请求（令牌桶）
# - 返回：是否允许通过 + 如果不行等多久

class RateLimiter:
    def check(self, user_id: str, tokens: int) -> tuple[bool, float]:
        # True + 0.0 = 允许
        # False + X.X = 需要等待 X.X 秒
        pass

---

Week 7：Capstone 项目（最重要的一周）

项目名：AgentKit — 生产级 Python Agent 框架

这是你的简历主打项目，面试时能完整讲清楚每个设计决策。

功能规格（必须全部完成）

core/
  agent.py          ← Agentic Loop（流式 + 取消 + 重试）
  tools/
    bash.py         ← BashTool（安全过滤 + 超时）
    file_read.py    ← FileReadTool（行号范围 + 大文件保护）
    file_write.py   ← FileWriteTool（原子写入）
    web_search.py   ← WebSearchTool（调 Serper/Bing API）
  mcp/
    server.py       ← MCP Server（stdio transport）
    client.py       ← MCP Client（进程管理 + 重连）
  session/
    store.py        ← JSONL 持久化（原子追加 + compact_boundary）
    restore.py      ← Session 恢复
  context/
    compressor.py   ← 自动上下文压缩（85% 阈值）
    tokens.py       ← Token 计数（离线估算 + 在线精确）
  cost/
    tracker.py      ← 成本统计（含 Prompt Cache 节省计算）
  permissions/
    checker.py      ← 三态权限（allow/deny/ask）
  retry/
    backoff.py      ← 指数退避重试
cli/
  main.py           ← 命令行入口（argparse）
  repl.py           ← 交互式 REPL
tests/
  fixtures/         ← VCR 录制文件
  test_agent.py
  test_tools.py
  test_mcp.py

每天进度

天	任务
Day 1	搭目录结构，实现 core/agent.py（集成 Week 1–3 成果）
Day 2	集成 tools/（4 个工具），接 MCP Server/Client
Day 3	实现 session/（持久化 + 恢复），接入 Agent
Day 4	实现 context/compressor.py + cost/tracker.py
Day 5	实现 permissions/ + retry/，完善错误处理
Day 6	CLI + REPL，写 README（说清楚架构图 + 设计决策）
Day 7	写测试（VCR fixture），录制 demo，推送 GitHub

README 必须包含的内容

• 架构图（用 ASCII 或 Mermaid）
• 每个核心模块的设计决策（为什么这样设计，有哪些 trade-off）
• 对比主流框架（LangChain/LangGraph）的区别

---

Week 8：面试冲刺

Day 1–2：系统设计训练

下面 5 道题，每道 20 分钟，用纸和笔画图 + 口述：

1. 设计一个生产级 Agentic Loop

   • 高并发（1000 并发 session）
   • 工具超时不阻塞其他用户
   • 失败重试不丢失上下文

2. 设计一个 Agent 工具市场

   • 开发者发布工具
   • Agent 运行时动态加载工具
   • 工具隔离（沙箱）

3. 设计一个多模型 Agent 系统

   • Orchestrator 用 Opus（复杂推理）
   • Worker 用 Haiku（批量处理）
   • 按任务类型路由

4. 设计上下文管理系统

   • 200k 窗口管理
   • 自动压缩 + 恢复
   • 多用户数据隔离

5. 设计 MCP 工具市场

   • 工具发现（类似 npm search）
   • 版本管理
   • 权限控制（哪些 Agent 能用哪些工具）

Day 3–4：面试题口述演练

下面每道题，对着镜子或录音，流畅口述 2–3 分钟（不看笔记）：

Agentic Loop 类

• Q：完整描述 Agentic Loop 的执行过程
• Q：工具并行执行是如何协调的？
• Q：如何检测 Agent 陷入循环？
• Q：Loop 中断恢复（用户 Ctrl+C 后重新连接）怎么做？

Context Management 类

• Q：上下文窗口满了怎么处理？如何选择丢弃哪些历史？
• Q：Prompt Cache 如何工作？如何最大化缓存命中率？
• Q：Token 计数有哪些方法？各自的精度和成本？

MCP 类

• Q：MCP 协议解决了什么问题？和 Function Calling 有什么区别？
• Q：设计一个支持 1000 个并发 MCP 客户端的服务器
• Q：MCP Server 崩溃时，Agent 如何处理？

安全类

• Q：如何防止 Prompt Injection 攻击？
• Q：BashTool 如何防止危险命令？列举 5 种攻击模式
• Q：工具的最小权限原则如何实现？

工程类

• Q：如何测试一个 LLM Agent？非确定性怎么处理？
• Q：Agent 系统的关键监控指标有哪些？
• Q：如何设计 Agent 系统的分布式追踪？

Day 5：Capstone 项目演练

用 10 分钟，完整介绍你的 AgentKit 项目：

• 解决了什么问题（1 分钟）
• 核心架构（3 分钟，画图）
• 最有挑战性的设计决策（3 分钟）
• 遇到的 bug 和如何解决（2 分钟）
• 和现有框架的对比（1 分钟）

Day 6：速记冲刺

• 通读 docs/面试速记口诀手册.md 2 遍（约 1 小时）
• 闭眼口述：Agentic Loop 7 步 / 权限 3 层 / SystemPrompt 5 层 / RLHF 3 步 / MoE 4 字
• 手写（不看文档）：LRU + 令牌桶 + 滑动窗口（各 10 分钟）

Day 7：针对性补漏

根据你的薄弱点，从以下选择：

• 算法偏弱 → 再刷 5 道 LeetCode（Medium，聚焦 HashMap / Heap / Sliding Window）
• 系统设计偏弱 → 看《设计数据密集型应用》第 1、2 章
• LLM 原理偏弱 → 读 docs/LLM技术深度系列/ 的 01（Transformer）和 03（RLHF）
• Python 语法偏弱 → 重读 docs/tech-stack/python/syntax/02-进阶特性.md（元类/描述符/dataclass）
• TypeScript 类型读不懂 → 重读 docs/tech-stack/typescript/syntax/02-进阶特性.md（条件类型/infer）
• MCP 偏弱 → 把你的 MCP Server 扩展到支持 10 个工具

---

必须手写的代码清单（验收标准）

两个月结束时，下面每项在 20 分钟内不看文档独立完成：

必须会的（核心，面试直接考）

• 完整 Agentic Loop（流式 + Tool Use + 重试 + 取消）
• 指数退避 with_retry（含 429/529 区分 + Retry-After 头）
• LRU Cache（双向链表实现，O(1)）
• 令牌桶限流器（O(1)，支持突发）
• 滑动窗口限流（精确版，O(N)）
• MCP Server（stdio transport，3 个以上工具）
• BashTool（含安全过滤 + 超时）
• JSONL Session 持久化（原子追加 + compact_boundary 恢复）
• 三态权限系统（allow / deny / ask + 通配符匹配）
• Top-K（最小堆实现）

应该会的（有可能考到）

• RAG Pipeline（chunking + embedding + 向量查询 + 混合检索）
• Sub-Agent（独立上下文 + 工具限制 + 级联取消）
• Prompt Cache 分段策略（为什么这几个位置放 breakpoint）
• 自动上下文压缩（阈值触发 + LLM 摘要 + compact_boundary）
• CircularBuffer（环形缓冲区，O(1) 读写）
• 一致性哈希（虚拟节点版）

---

面试当天注意事项

1. 开口第一句话：描述你的 Capstone 项目，建立技术可信度
2. 系统设计：先说假设（并发量、数据量），再画架构图，从简单到复杂
3. 算法题：先说思路（时间复杂度），再写代码，写完说优化方向
4. 不会的题：说”我没有直接经验，但我会这样思考…”，展示方法论
5. 源码引用：说”Claude Code 的 withRetry 用了这个设计，原因是…”，展示深度

---

进度追踪表

复制到你的笔记本，每周末更新：

Week 0  □ Python语法自检  □ LLM理论自检  □ TS可读性自检  （不需要则跳过）
Week 1  □ asyncio 四练  □ Agentic Loop v1  □ withRetry  □ 取消机制  □ 验收通过
Week 2  □ Tool 基类     □ BashTool         □ 权限系统   □ 路径安全  □ 验收通过
Week 3  □ MCP Server    □ MCP Client       □ 集成Loop   □ SSE传输   □ 验收通过
Week 4  □ JSONL存储     □ Session恢复      □ SystemPrompt □ 自动压缩 □ 验收通过
Week 5  □ Sub-Agent     □ Fork模式         □ Coordinator □ 错误隔离 □ 验收通过
Week 6  □ RAG           □ 混合检索         □ 算法5道   □ 算法复写  □ 验收通过
Week 7  □ Capstone框架  □ 全功能集成       □ CLI        □ 测试+文档 □ GitHub推送
Week 8  □ 系统设计5道   □ 面试题口述       □ 项目演练  □ 补漏      □ 准备就绪